//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by FernFlower decompiler)
//

package com.ververica.cdc.connectors.kingbase.table;

import com.ververica.cdc.connectors.base.options.JdbcSourceOptions;
import com.ververica.cdc.connectors.base.options.SourceOptions;
import com.ververica.cdc.connectors.base.options.StartupMode;
import com.ververica.cdc.connectors.base.options.StartupOptions;
import com.ververica.cdc.connectors.base.utils.ObjectUtils;
import com.ververica.cdc.connectors.postgres.source.config.PostgresSourceOptions;
import com.ververica.cdc.connectors.postgres.table.PostgreSQLTableSource;
import com.ververica.cdc.connectors.postgres.utils.OptionUtils;
import com.ververica.cdc.debezium.table.DebeziumChangelogMode;
import com.ververica.cdc.debezium.table.DebeziumOptions;
import com.ververica.cdc.debezium.utils.ResolvedSchemaUtils;
import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.util.Preconditions;

public class PostgreSQLTableFactory implements DynamicTableSourceFactory {
    private static final String IDENTIFIER = "kingbase-cdc";
    private static final String SCAN_STARTUP_MODE_VALUE_INITIAL = "initial";
    private static final String SCAN_STARTUP_MODE_VALUE_LATEST = "latest-offset";

    public PostgreSQLTableFactory() {
    }

    public DynamicTableSource createDynamicTableSource(DynamicTableFactory.Context context) {
        FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
        helper.validateExcept(new String[]{"debezium."});
        ReadableConfig config = helper.getOptions();
        String hostname = (String)config.get(JdbcSourceOptions.HOSTNAME);
        String username = (String)config.get(JdbcSourceOptions.USERNAME);
        String password = (String)config.get(JdbcSourceOptions.PASSWORD);
        String databaseName = (String)config.get(JdbcSourceOptions.DATABASE_NAME);
        String schemaName = (String)config.get(JdbcSourceOptions.SCHEMA_NAME);
        String tableName = (String)config.get(JdbcSourceOptions.TABLE_NAME);
        int port = (Integer)config.get(PostgresSourceOptions.PG_PORT);
        String pluginName = (String)config.get(PostgresSourceOptions.DECODING_PLUGIN_NAME);
        String slotName = (String)config.get(PostgresSourceOptions.SLOT_NAME);
        DebeziumChangelogMode changelogMode = (DebeziumChangelogMode)config.get(PostgresSourceOptions.CHANGELOG_MODE);
        ResolvedSchema physicalSchema = ResolvedSchemaUtils.getPhysicalSchema(context.getCatalogTable().getResolvedSchema());
        if (changelogMode == DebeziumChangelogMode.UPSERT) {
            Preconditions.checkArgument(physicalSchema.getPrimaryKey().isPresent(), "Primary key must be present when upsert mode is selected.");
        }

        boolean enableParallelRead = (Boolean)config.get(PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED);
        StartupOptions startupOptions = getStartupOptions(config);
        int splitSize = (Integer)config.get(PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE);
        int splitMetaGroupSize = (Integer)config.get(PostgresSourceOptions.CHUNK_META_GROUP_SIZE);
        int fetchSize = (Integer)config.get(PostgresSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE);
        Duration connectTimeout = (Duration)config.get(PostgresSourceOptions.CONNECT_TIMEOUT);
        int connectMaxRetries = (Integer)config.get(PostgresSourceOptions.CONNECT_MAX_RETRIES);
        int connectionPoolSize = (Integer)config.get(PostgresSourceOptions.CONNECTION_POOL_SIZE);
        double distributionFactorUpper = (Double)config.get(PostgresSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND);
        double distributionFactorLower = (Double)config.get(PostgresSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
        Duration heartbeatInterval = (Duration)config.get(PostgresSourceOptions.HEARTBEAT_INTERVAL);
        String chunkKeyColumn = (String)config.getOptional(PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN).orElse(null);
        boolean closeIdlerReaders = (Boolean)config.get(SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
        boolean skipSnapshotBackfill = (Boolean)config.get(SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
        if (enableParallelRead) {
            this.validateIntegerOption(PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1);
            this.validateIntegerOption(PostgresSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE, fetchSize, 1);
            this.validateIntegerOption(PostgresSourceOptions.CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1);
            this.validateIntegerOption(JdbcSourceOptions.CONNECTION_POOL_SIZE, connectionPoolSize, 1);
            this.validateIntegerOption(JdbcSourceOptions.CONNECT_MAX_RETRIES, connectMaxRetries, 0);
            this.validateDistributionFactorUpper(distributionFactorUpper);
            this.validateDistributionFactorLower(distributionFactorLower);
        } else {
            Preconditions.checkState(!StartupMode.LATEST_OFFSET.equals(startupOptions.startupMode), "The Postgres CDC connector does not support 'latest-offset' startup mode when 'scan.incremental.snapshot.enabled' is disabled, you can enable 'scan.incremental.snapshot.enabled' to use this startup mode.");
        }

        OptionUtils.printOptions("kingbase-cdc", ((Configuration)config).toMap());
        return new PostgreSQLTableSource(physicalSchema, port, hostname, databaseName, schemaName, tableName, username, password, pluginName, slotName, changelogMode, DebeziumOptions.getDebeziumProperties(context.getCatalogTable().getOptions()), enableParallelRead, splitSize, splitMetaGroupSize, fetchSize, connectTimeout, connectMaxRetries, connectionPoolSize, distributionFactorUpper, distributionFactorLower, heartbeatInterval, startupOptions, chunkKeyColumn, closeIdlerReaders, skipSnapshotBackfill);
    }

    public String factoryIdentifier() {
        return "kingbase-cdc";
    }

    public Set<ConfigOption<?>> requiredOptions() {
        Set<ConfigOption<?>> options = new HashSet();
        options.add(JdbcSourceOptions.HOSTNAME);
        options.add(JdbcSourceOptions.USERNAME);
        options.add(JdbcSourceOptions.PASSWORD);
        options.add(JdbcSourceOptions.DATABASE_NAME);
        options.add(JdbcSourceOptions.SCHEMA_NAME);
        options.add(JdbcSourceOptions.TABLE_NAME);
        options.add(PostgresSourceOptions.SLOT_NAME);
        return options;
    }

    public Set<ConfigOption<?>> optionalOptions() {
        Set<ConfigOption<?>> options = new HashSet();
        options.add(PostgresSourceOptions.PG_PORT);
        options.add(PostgresSourceOptions.DECODING_PLUGIN_NAME);
        options.add(PostgresSourceOptions.CHANGELOG_MODE);
        options.add(PostgresSourceOptions.SCAN_STARTUP_MODE);
        options.add(PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED);
        options.add(PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE);
        options.add(PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN);
        options.add(PostgresSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND);
        options.add(PostgresSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND);
        options.add(PostgresSourceOptions.CHUNK_META_GROUP_SIZE);
        options.add(PostgresSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE);
        options.add(PostgresSourceOptions.CONNECT_TIMEOUT);
        options.add(PostgresSourceOptions.CONNECT_MAX_RETRIES);
        options.add(PostgresSourceOptions.CONNECTION_POOL_SIZE);
        options.add(PostgresSourceOptions.HEARTBEAT_INTERVAL);
        options.add(SourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED);
        options.add(SourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP);
        return options;
    }

    private static StartupOptions getStartupOptions(ReadableConfig config) {
        String modeString = (String)config.get(PostgresSourceOptions.SCAN_STARTUP_MODE);
        switch (modeString.toLowerCase()) {
            case "initial":
                return StartupOptions.initial();
            case "latest-offset":
                return StartupOptions.latest();
            default:
                throw new ValidationException(String.format("Invalid value for option '%s'. Supported values are [%s, %s], but was: %s", PostgresSourceOptions.SCAN_STARTUP_MODE.key(), "initial", "latest-offset", modeString));
        }
    }

    private void validateIntegerOption(ConfigOption<Integer> option, int optionValue, int exclusiveMin) {
        Preconditions.checkState(optionValue > exclusiveMin, String.format("The value of option '%s' must larger than %d, but is %d", option.key(), exclusiveMin, optionValue));
    }

    private void validateDistributionFactorUpper(double distributionFactorUpper) {
        Preconditions.checkState(ObjectUtils.doubleCompare(distributionFactorUpper, 1.0) >= 0, String.format("The value of option '%s' must larger than or equals %s, but is %s", PostgresSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND.key(), 1.0, distributionFactorUpper));
    }

    private void validateDistributionFactorLower(double distributionFactorLower) {
        Preconditions.checkState(ObjectUtils.doubleCompare(distributionFactorLower, 0.0) >= 0 && ObjectUtils.doubleCompare(distributionFactorLower, 1.0) <= 0, String.format("The value of option '%s' must between %s and %s inclusively, but is %s", PostgresSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND.key(), 0.0, 1.0, distributionFactorLower));
    }
}
